package com.miya.demo.support.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

/**
 * kafka消费者handler
 *
 * @author CaiXiaowei
 * @date 2023/6/2
 */
@Component
@Slf4j
public class KafkaConsumerExceptionHandler implements KafkaListenerErrorHandler {


    /**
     * @param message
     * @param e
     * @return
     */
    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException e) {
        log.error("kafka消费异常, message:{}, error:{}", message, e);
        return null;
    }

    /**
     * @param message
     * @param exception
     * @param consumer
     * @return
     */
    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
        log.error("消息详情: {}" + message);
        log.error("异常信息: {}" + exception);
        log.error("消费者详情: {}" + consumer.groupMetadata());
        log.error("监听主题: {}" + consumer.listTopics());
        return KafkaListenerErrorHandler.super.handleError(message, exception, consumer);
    }
}
